package teacher

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object LogDeal {
  def main(args: Array[String]): Unit = {
    /**
      *
      */
    if(args.length != 2) {
      println(
        """
          |cn.dmp.etl.Bzip2Parquet
          |params:
          | dataInputPath 日志输入路径
          | outputPath  结果输出路径
        """.stripMargin)
      sys.exit()
    }

    //        val dataInputPath = args(0)
    //        val outputPath = args(1)

    val Array(dataInputPath, outputPath) = args


    // sparkcontext
    val sparkConf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName}")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val sQLContext = new SQLContext(sc)
    sQLContext.setConf("spark.sql.parquet.compression.codec", "snappy")

    // 读取数据
    val rawLog = sc.textFile(dataInputPath)

    // 处理数据 a,b,c,,,
    val filtered = rawLog.map(t => t.split(",", t.length)).filter(_.length >= 85)

    import xubo.wangcaifeng.love.Utils.XuboString._
    val muchRow: RDD[Row] = filtered.map(arr => {
      Row(
        arr(0),
        arr(1).toIntPlus,
        arr(2).toIntPlus,
        arr(3).toIntPlus,
        arr(4).toIntPlus,
        arr(5),
        arr(6),
        arr(7).toIntPlus,
        arr(8).toIntPlus,
        arr(9).toDoublePlus,
        arr(10).toDoublePlus,
        arr(11),
        arr(12),
        arr(13),
        arr(14),
        arr(15),
        arr(16),
        arr(17).toIntPlus,
        arr(18),
        arr(19),
        arr(20).toIntPlus,
        arr(21).toIntPlus,
        arr(22),
        arr(23),
        arr(24),
        arr(25),
        arr(26).toIntPlus,
        arr(27),
        arr(28).toIntPlus,
        arr(29),
        arr(30).toIntPlus,
        arr(31).toIntPlus,
        arr(32).toIntPlus,
        arr(33),
        arr(34).toIntPlus,
        arr(35).toIntPlus,
        arr(36).toIntPlus,
        arr(37),
        arr(38).toIntPlus,
        arr(39).toIntPlus,
        arr(40).toDoublePlus,
        arr(41).toDoublePlus,
        arr(42).toIntPlus,
        arr(43),
        arr(44).toDoublePlus,
        arr(45).toDoublePlus,
        arr(46),
        arr(47),
        arr(48),
        arr(49),
        arr(50),
        arr(51),
        arr(52),
        arr(53),
        arr(54),
        arr(55),
        arr(56),
        arr(57).toIntPlus,
        arr(58).toDoublePlus,
        arr(59).toIntPlus,
        arr(60).toIntPlus,
        arr(61),
        arr(62),
        arr(63),
        arr(64),
        arr(65),
        arr(66),
        arr(67),
        arr(68),
        arr(69),
        arr(70),
        arr(71),
        arr(72),
        arr(73).toIntPlus,
        arr(74).toDoublePlus,
        arr(75).toDoublePlus,
        arr(76).toDoublePlus,
        arr(77).toDoublePlus,
        arr(78).toDoublePlus,
        arr(79),
        arr(80),
        arr(81),
        arr(82),
        arr(83),
        arr(84).toIntPlus
      )
    })

    // 描述信息
    val schema = StructType(Seq(
      StructField("sessionid", StringType),
      StructField("advertisersid", IntegerType),
      StructField("adorderid", IntegerType),
      StructField("adcreativeid", IntegerType),
      StructField("adplatformproviderid", IntegerType),
      StructField("sdkversion", StringType),
      StructField("adplatformkey", StringType),
      StructField("putinmodeltype", IntegerType),
      StructField("requestmode", IntegerType),
      StructField("adprice", DoubleType),
      StructField("adppprice", DoubleType),
      StructField("requestdate", StringType),
      StructField("ip", StringType),
      StructField("appid", StringType),
      StructField("appname", StringType),
      StructField("uuid", StringType),
      StructField("device", StringType),
      StructField("client", IntegerType),
      StructField("osversion", StringType),
      StructField("density", StringType),
      StructField("pw", IntegerType),
      StructField("ph", IntegerType),
      StructField("long", StringType),
      StructField("lat", StringType),
      StructField("provincename", StringType),
      StructField("cityname", StringType),
      StructField("ispid", IntegerType),
      StructField("ispname", StringType),
      StructField("networkmannerid", IntegerType),
      StructField("networkmannername", StringType),
      StructField("iseffective", IntegerType),
      StructField("isbilling", IntegerType),
      StructField("adspacetype", IntegerType),
      StructField("adspacetypename", StringType),
      StructField("devicetype", IntegerType),
      StructField("processnode", IntegerType),
      StructField("apptype", IntegerType),
      StructField("district", StringType),
      StructField("paymode", IntegerType),
      StructField("isbid", IntegerType),
      StructField("bidprice", DoubleType),
      StructField("winprice", DoubleType),
      StructField("iswin", IntegerType),
      StructField("cur", StringType),
      StructField("rate", DoubleType),
      StructField("cnywinprice", DoubleType),
      StructField("imei", StringType),
      StructField("mac", StringType),
      StructField("idfa", StringType),
      StructField("openudid", StringType),
      StructField("androidid", StringType),
      StructField("rtbprovince", StringType),
      StructField("rtbcity", StringType),
      StructField("rtbdistrict", StringType),
      StructField("rtbstreet", StringType),
      StructField("storeurl", StringType),
      StructField("realip", StringType),
      StructField("isqualityapp", IntegerType),
      StructField("bidfloor", DoubleType),
      StructField("aw", IntegerType),
      StructField("ah", IntegerType),
      StructField("imeimd5", StringType),
      StructField("macmd5", StringType),
      StructField("idfamd5", StringType),
      StructField("openudidmd5", StringType),
      StructField("androididmd5", StringType),
      StructField("imeisha1", StringType),
      StructField("macsha1", StringType),
      StructField("idfasha1", StringType),
      StructField("openudidsha1", StringType),
      StructField("androididsha1", StringType),
      StructField("uuidunknow", StringType),
      StructField("userid", StringType),
      StructField("iptype", IntegerType),
      StructField("initbidprice", DoubleType),
      StructField("adpayment", DoubleType),
      StructField("agentrate", DoubleType),
      StructField("lomarkrate", DoubleType),
      StructField("adxrate", DoubleType),
      StructField("title", StringType),
      StructField("keywords", StringType),
      StructField("tagid", StringType),
      StructField("callbackdate", StringType),
      StructField("channelid", StringType),
      StructField("mediatype", IntegerType)
    ))


    val dataFrame = sQLContext.createDataFrame(muchRow, schema)

    // 存储数据 repartition 重新分区    coalesce 合并分区默认没有shuffle
    dataFrame/*.repartition(4)*//*.coalesce(2)*/.write.parquet(outputPath)
    //

    // 关闭sparkcontext
    sc.stop()
  }

}
